package network.networkProfile;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import network.SendBatch;

/**
 * stores information related to the state of the connection between localhost
 * and some target host, used to predict packet dropping, responsible for managing
 * packet ackowledgment
 * <br><br>
 * this class uses a timeout to determine packet loss and this must
 * be continually updated in order to function correctly
 * <br><br>
 * this class relies on a steady stream of packets to and from the associated network
 * address and the host; packet ack information can be sent in reliable or unreliable
 * packets, reliable only means that the network profile wants an ack back for the
 * received packet
 * <br><br>
 * improvements to make:
 * instead of storing old packets, store the discrete data snippets (the kind that come in
 * the send batches) and which acks were written to the header (if any); this allows for
 * sending time sensitive packet information; ie, if a packet is reliable but is lost, instead
 * of resendnig the whole packet, check first to see if any of the information contained was
 * time sensitive and throw out expired data; acks that are known to have expired can also be
 * thrown out this way if the original packet id is not used when a new packet is built and resent
 * 
 * 
 * @author jack
 *
 */
public class NetworkProfile
{
	/**
	 * the max size of packets in bytes
	 */
	private static final short maxPacketSize = 1024;
	/**
	 * if true, packets are packed until the are at or over the max packet
	 * size; else the packets are made to be less than or equal to the max packet size
	 */
	private static final boolean roughSizing = false;
	/**
	 * the maximum number of acks written to a packet header,
	 * unless the noAckLimit parameter is set to true, if there are
	 * more pending acks than this number, the send() method is called
	 * (which will send out ack only packets in addition to unsent lost packets)
	 */
	private static final int maxAcksPerPacket = 30;
	
	private DatagramSocket s;
	private InetAddress address;
	private int port;
	private short reliablePacketIDindex = Short.MIN_VALUE;
	private long unreliablePacketIDindex = Long.MIN_VALUE;
	private int lostPackets = 0; //number of packets that failed the timeout
	private int sentPackets = 0; //each burst of packets is counted as a single 'packet' sent
	private int packetsSinceLoss = 0; //number of packets ackownledged since the last packet loss occured, used to predict drop probability
	
	/**
	 * ids of received packets, ids are removed after a (long) timeout
	 */
	private HashSet<Short> receivedPacketIDs = new HashSet<Short>();
	/**
	 * blocking queue of received packet ids in order to remove them from the set
	 * of received ids after a timeout
	 */
	private LinkedBlockingQueue<PacketRecord> receivedPackets = new LinkedBlockingQueue<PacketRecord>();
	/**
	 * the timeout for received packets before they are cleared from the profile
	 */
	private long receivedPacketTimeout = 15000;
	/**
	 * pending ackowledgements that need to be sent to the associated address
	 */
	private LinkedBlockingQueue<Short> pendingAcks = new LinkedBlockingQueue<Short>();
	/**
	 * reliable packets that were sent to the associated network address, stored
	 * for resending if they timeout
	 */
	private Map<Short, byte[]> storedReliablePackets = Collections.synchronizedMap(new HashMap<Short, byte[]>());
	/**
	 * time before a packet declared lost; if an ack is received from the associated
	 * profile before this time, then the packet is declared to have arrived
	 */
	private long timeout = 2000;
	/**
	 * packets that timed out and need to be resent
	 */
	private LinkedBlockingQueue<Short> resendPackets = new LinkedBlockingQueue<Short>();
	/**
	 * records the time since the last header was sent, -1 is default starting value,
	 * this time is updated every time a reliable packet is sent, or if its value
	 * is the default value and reliable packet is received
	 */
	private long headerSendTime = -1;
	/**
	 * max time between header sends, if headerSendTime is greater then this amount, the
	 * send method will be called and all reliable packets waiting for resend will be sent
	 * and a packet containing header information (acks, etc) will also be sent reliably
	 */
	private long headerSendTimeout = 100;
	/**
	 * records the packets that have been sent for which no ack has been received;
	 * the information is stored in a list because the timeout for all the packets is
	 * the same; hence, to check if a packet is past the timeout (and thusly declared lost)
	 * we need only start at the beginning of the queue and check the send times (to store
	 * this data we use simple class PacketRecord)
	 * <br><br>
	 * packets the need to be resent are transfered out of the waiting queue and into the
	 * resend packets queue
	 * <br><br>
	 * this object should be thread safe because it implements BlockingQueue
	 * see http://download.oracle.com/javase/1,5.0/docs/api/java/util/concurrent/BlockingQueue.html
	 */
	private LinkedBlockingQueue<PacketRecord> waitingPackets;
	/**
	 * packets that have been successfully ackowledged 
	 */
	private Set<Short> ackedPackets = Collections.synchronizedSet(new HashSet<Short>());
	private class PacketRecord
	{
		private short id;
		private long time;
		public PacketRecord(short id, long time)
		{
			this.id = id;
			this.time = time;
		}
		public short getID()
		{
			return id;
		}
		public long getTime()
		{
			return time;
		}
	}
	/**
	 * creates a new network profile
	 * @param socket the socket to use to send the datagrams
	 * @param address the network address this profile is associated with
	 * @param port the port the profile sends packets to
	 * @throws IOException
	 */
	public NetworkProfile(DatagramSocket socket, InetAddress address, int port) throws IOException
	{
		s = socket;
		this.address = address;
		this.port = port;
		waitingPackets = new LinkedBlockingQueue<PacketRecord>();
	}
	/**
	 * updates the network profile, used to determine timeouts for
	 * packets; this method should be extremely cheap to call
	 */
	public void update()
	{
		//System.out.println("updating...");
		long time = System.currentTimeMillis();
		while(waitingPackets.size() > 0 && time - waitingPackets.peek().getTime() >= timeout || ackedPackets.contains(waitingPackets.peek()))
		{
			if(ackedPackets.contains(waitingPackets.peek().getID()))
			{
				System.out.println("waiting packet "+waitingPackets.peek().getID()+" acked! removing from packet store");
				storedReliablePackets.remove(waitingPackets.poll().getID());
			}
			else
			{
				System.out.println("waiting packet "+waitingPackets.peek().getID()+" declared lost (time ~= "+(time - waitingPackets.peek().getTime())+"), adding to resend queue");
				lostPackets++;
				packetsSinceLoss = 0;
				resendPackets.add(waitingPackets.poll().getID());
			}
		}
		while(receivedPackets.size() > 0 && time - receivedPackets.peek().getTime() >= receivedPacketTimeout)
		{
			//merely clears the cache of received packet ids so they dont build up,
			//allows the early ids to be reused if need be
			receivedPacketIDs.remove(receivedPackets.poll().getID());
		}
		if(headerSendTime != -1 && System.currentTimeMillis() - headerSendTime >= headerSendTimeout || pendingAcks.size() > maxAcksPerPacket)
		{
			try
			{
				send();
			}
			catch(IOException e)
			{
				e.printStackTrace();
			}
		}
	}
	/**
	 * called to construct a header to append to the beginning of an
	 * outgoing packet directed towards the address associated with
	 * this profile
	 * 
	 * this method increments the outgoing packet id index for this profile
	 * and records this as a packet sent
	 * 
	 * @param reliable determines whether the header should be formulated for
	 * a reliable or unreliable packet
	 * @param sendAcks determines whether acks should be included in the packet or not
	 * @param = noAckLimit if true there is no limit to the amount of acks placed in
	 * a single packet (other than max packet size), else acks are limited to use only
	 * the 30 bytes (15 acks of 2 byte short ids)
	 * @return returns a header
	 */
	private byte[] getHeader(boolean reliable, boolean sendAcks, boolean noAckLimit) throws IOException
	{
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		DataOutputStream dos = new DataOutputStream(baos);

		dos.writeBoolean(reliable);
		if(reliable)
		{
			dos.writeShort(reliablePacketIDindex);
			if(ackedPackets.contains(reliablePacketIDindex))
			{
				//starts removing old acked packet ids once the ids roll over
				ackedPackets.remove(reliablePacketIDindex);
				//System.out.println("REMOVED OLD ACK FROM SET");
			}
			
			//sentPackets++; this is updated when the packet is sent
			reliablePacketIDindex++;
			if(reliablePacketIDindex == Short.MAX_VALUE)
			{
				reliablePacketIDindex = Short.MIN_VALUE;
			}
		}
		else
		{
			dos.writeLong(unreliablePacketIDindex);
			unreliablePacketIDindex++;
		}
		if(sendAcks)
		{
			//dumps the acks
			int acksToDump = pendingAcks.size();
			ByteArrayOutputStream ackbaos = new ByteArrayOutputStream();
			DataOutputStream ackdos = new DataOutputStream(ackbaos);
			System.out.print("adding acks: ");
			int i = 0;
			for(;i < acksToDump && (noAckLimit || ackbaos.size() < maxAcksPerPacket) && ackbaos.size()+20 < NetworkProfile.maxPacketSize; i++)
			{
				System.out.print(pendingAcks.peek()+", ");
				ackdos.writeShort(pendingAcks.poll());
			}
			System.out.println();
			dos.writeBoolean(true);
			dos.write(Byte.MIN_VALUE+i);
			dos.write(ackbaos.toByteArray());
		}
		else
		{
			dos.writeBoolean(false);
		}
		return baos.toByteArray();
	}
	/**
	 * determines the number of packets to send in order to achieve at
	 * or above the specified probability of success
	 * @param probSuccess
	 * @return returns the number of packets to send
	 */
	private int getPacketsToSend(double probSuccess)
	{
		//  (1-probSuc)^n <= dropProb  ==>  log(dropProb)/log(1-probSuc) ~ n
		//int packets = (int)Math.round(Math.log(1-probSuccess)/Math.log(predictPacketLoss()));
		//System.out.println(Math.log(predictPacketLoss())/Math.log(1-probSuccess));
		int packets = (int)Math.round(Math.log(predictPacketLoss())/Math.log(1-probSuccess));
		return packets <= 0? 1: packets;
	}
	/**
	 * sends a batch of data to connected clients
	 * @param batch
	 */
	public void send(SendBatch batch) throws IOException
	{
		System.out.println("send, batch = "+batch);
		//dumps the timed out reliable packets to be resent in this round
		dumpResendPackets();
		if(batch.getReliableData().size() > 0)
		{
			headerSendTime = System.currentTimeMillis();
			packSend(batch.getReliableData(), true, true);
		}
		if(batch.getUnreliableData().size() > 0)
			packSend(batch.getUnreliableData(), false, false);
	}
	/**
	 * sends just header information and packets that have been deemed lost,
	 * this method gets called if either a reliable packet has not been sent
	 * in a while or if the send method has not been called in a while and there
	 * are reliable packets waiting for resend
	 * 
	 * header only packets do not count towards packets sent nor do they require acks
	 * @throws IOException
	 */
	private void send() throws IOException
	{
		headerSendTime = System.currentTimeMillis();
		//System.out.println("send, no batch, pending acks = "+pendingAcks.size());
		if(pendingAcks.size() > 0)
		{
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			DataOutputStream dos = new DataOutputStream(baos);
			byte[] header = getHeader(false, true, true);
			//dos.write((byte)(Byte.MIN_VALUE+header.length));
			dos.write(header);
			byte[] temp = baos.toByteArray();
			DatagramPacket p = new DatagramPacket(temp, temp.length, address, port);
			sendPacket(p, .99, false); //sent reliably but without a need for ackowledgement once received
		}
		
		//dumps the timed out reliable packets to be resent in this round
		dumpResendPackets();
	}
	/**
	 * dumps that old reliable packets that should be resent to the associated address
	 */
	private void dumpResendPackets() throws IOException
	{
		while(resendPackets.size() > 0)
		{
			//System.out.println("dumping resend packet, "+resendPackets.peek());
			short packetID = resendPackets.poll();
			byte[] packetData = storedReliablePackets.get(packetID);
			//System.out.println("packet data = null, "+storedReliablePackets);
			if(packetData != null) //packet data might be removed before it gets resent if ack received
			{
				/*int packetsToSend = getPacketsToSend(.99);
				DatagramPacket p = new DatagramPacket(packetData, packetData.length, address, port);
				for(int i = 0; i < packetsToSend; i++)
				{
					s.send(p);
				}*/
				DatagramPacket p = new DatagramPacket(packetData, packetData.length, address, port);
				sendPacket(p, .99, true);
			}
			else
			{
				System.out.println("PACKET DATA = NULL, "+storedReliablePackets);
			}
		}
		headerSendTime = System.currentTimeMillis();
	}
	/**
	 * packs the passed data into packets and sends them to the associated network address,
	 * if the packet is marked as reliable it sends a burst of packets to attempt 99% chance
	 * to receive on client side, if the packet is unreliable the profile attemps to send with
	 * 50% chance of receiving
	 * @param data
	 * @param reliable
	 * @throws IOException
	 */
	private void packSend(ArrayList<byte[]> data, boolean reliable, boolean sendAcks) throws IOException
	{
		System.out.println("sending: "+data);
		while(data.size() > 0)
		{
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			DataOutputStream dos = new DataOutputStream(baos);
			byte[] header = getHeader(reliable, sendAcks, false);
			//dos.write((byte)(Byte.MIN_VALUE+header.length));
			dos.write(header);
			if(data != null)
			{
				for(Iterator<byte[]> i = data.iterator(); i.hasNext() && baos.size() <= NetworkProfile.maxPacketSize;)
				{
					byte[] b = i.next();
					if(b.length > NetworkProfile.maxPacketSize)
					{
						System.err.println("byte buffer over max size, length = "+b.length+", max packet size = "+NetworkProfile.maxPacketSize);
						i.remove();
					}
					else if(roughSizing || b.length+baos.size() <= NetworkProfile.maxPacketSize) //rough sizing allows the packet to go a little over
					{
						dos.write(b);
						i.remove();
					}
				}
			}
			byte[] temp = baos.toByteArray();

			DatagramPacket p = new DatagramPacket(temp, temp.length, address, port);
			if(reliable)
			{
				sendPacket(p, .99, reliable);
			}
			else
			{
				sendPacket(p, 50, reliable);
			}
		}
	}
	/**
	 * does the work of actually sending a packet to the associated network address
	 * @param chanceToRecieve the probability that the associated network
	 * address should receive the packet, used to determine the number of
	 * packets to be sent in a packet burst, this probability represents
	 * a minimum chance (the actual chance could be higher)
	 * @param reliable if packet is reliable it is added to waiting list
	 * to await an ack and packets sent is incremented
	 */
	private void sendPacket(DatagramPacket p, double chanceToRecieve, boolean reliable) throws IOException
	{
		System.out.println("packet sent, reliable = "+reliable);
		//formulated this way for efficiency reasons (expensive to create 'for' loop)
		int packetsToSend = getPacketsToSend(.99);
		if(packetsToSend == 1)
		{
			s.send(p);
		}
		else
		{
			for(int i = 0; i < packetsToSend; i++)
			{
				s.send(p);
			}
		}
		if(reliable)
		{
			//note: necessary because packets can be resent, and thus might already be stored in the packet store
			ByteBuffer temp = ByteBuffer.wrap(p.getData(), 1, 3);
			short id = temp.getShort();
			//System.out.println("adding to packet store, id = "+id);
			if(!storedReliablePackets.containsKey(id))
				storedReliablePackets.put(id, p.getData());
			waitingPackets.add(new PacketRecord(id, System.currentTimeMillis()));
			sentPackets++;
			packetsSinceLoss++;
		}
	}
	/**
	 * calculates the predicted drop rate for this network profile based off
	 * the statistics for dropped and sent packets
	 * 
	 * for the purposes of calculation, waiting packets (packets whose ack has
	 * not been received) are counted as arrived packets (algorithm is optimistic)
	 * 
	 * see http://en.wikipedia.org/wiki/Bayesian_inference for more information
	 * @return returns the predicted packet loss percent
	 */
	private double predictPacketLoss()
	{
		//P(H)
		double pH = lostPackets*1. / sentPackets; //prior probability, inferred before the evidence is aquired
		
		//P(E|H) probability for event E (the evidence) to occur given prob. H for its occurence
		//E should be something related to packets being dropped or received
		//ie, if E = the last 2 packets were ack. then P(E|H) = (1-P(H))^2
		
		//evidence used is the number of packets ack. before packet loss
		double pEH = Math.pow(1-pH, packetsSinceLoss);
		
		//P(E)
		double pE = .03; //prior probability, chance for any packet to be lost, according to internet 95%-99% packets survive UDP
		//could have the server record this data from each profile when shutting down instead of guessing
		
		return pEH * pH / pE;
	}
	/**
	 * processes the header contained at the beginning of the packet received
	 * from the network address associated with this profile
	 * @param header the header of the received packet
	 * @return returns true if the information contained in the packet should be processed, false otherwise
	 */
	public boolean processHeader(ByteBuffer header)
	{
		System.out.println("processing header");
		boolean reliable = header.get()==1;
		if(reliable)
		{
			if(headerSendTime == -1)
			{
				System.out.println("\treceived first reliable packet, header send time set off default");
				headerSendTime = System.currentTimeMillis();
			}
			short packetID = header.getShort();
			System.out.println("\treliable packet, short id = "+packetID);
			//checks to see if this packet was already received (ie, its a burst), if not adds to pending acks
			if(!receivedPacketIDs.contains(packetID))
			{
				receivedPacketIDs.add(packetID);
				receivedPackets.add(new PacketRecord(packetID, System.currentTimeMillis()));
				pendingAcks.add(packetID);
				System.out.println("\tnew packet detected, added to pending ack list");
			}
			else
			{
				System.out.println("\tpacket "+packetID+" already received, assumed part of burst, not processing");
				return false;
			}
		}
		else
		{
			long packetID = header.getLong(); //not used currently
			System.out.println("\tunreliable packet, long id = "+packetID);
		}
		boolean containsAcks = header.get()==1;
		System.out.println("\tcontains acks = "+containsAcks);
		if(containsAcks)
		{
			System.out.println("\treading acks");
			int acks = header.get()-Byte.MIN_VALUE;
			for(int i = 0; i < acks; i++)
			{
				short ackID = header.getShort();
				System.out.println("\t\t"+ackID+" acked");
				//System.out.println("packet "+ackID+" acked!, ack complete!");
				//waitingPackets.remove(ackID); //removed in the update method when they come up, less expensive operation
				ackedPackets.add(ackID);
			}
		}
		return true;
	}
	public String toString()
	{
		return "predicted packet loss = "+predictPacketLoss()+", packets per burst = "+getPacketsToSend(.99)+
			", lost packets / sent packets = "+lostPackets+" / "+sentPackets+" = "+(lostPackets*1./sentPackets)+"\n"+
			"waiting packets = "+waitingPackets.size()+", pending acks = "+pendingAcks.size()+", packets since loss = "+packetsSinceLoss;
	}
}